package main

import (
	"encoding/json"
	"log"
	"net/http"
	"time"

	"github.com/IBM/sarama"
	"github.com/prebid/openrtb/v20/openrtb2"
)

var (
	kafkaBrokerList = []string{"172.31.29.134:9092"} // Kafka broker地址
	kafkaTopic      = "device-info"                  // Kafka主题
	batchSize       = 10000                          // 批量发送的大小
	batchTimeout    = 200 * time.Millisecond         // 批量发送的超时时间
	//num             = 0                          // 用于记录消息数量
)

type message struct {
	value []byte
}

func main() {
	// 初始化Kafka生产者
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.NoResponse
	config.Producer.Retry.Max = 10
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true

	producer, err := sarama.NewAsyncProducer(kafkaBrokerList, config)
	if err != nil {
		log.Fatalf("Failed to start Sarama async producer: %v", err)
	}
	defer func() {
		if err := producer.Close(); err != nil {
			log.Fatalf("Failed to close Sarama async producer: %v", err)
		}
	}()

	// 创建一个通道来接收消息
	messageChan := make(chan *message, batchSize)

	// 启动一个goroutine来批量发送消息
	go func() {
		var messages []*sarama.ProducerMessage
		ticker := time.NewTicker(batchTimeout)
		defer ticker.Stop()

		for {
			select {
			case msg := <-messageChan:
				messages = append(messages, &sarama.ProducerMessage{
					Topic: kafkaTopic,
					Value: sarama.ByteEncoder(msg.value),
				})

				if len(messages) >= batchSize {
					sendMessages(producer, messages)
					messages = nil
				}
			case <-ticker.C:
				if len(messages) > 0 {
					sendMessages(producer, messages)
					messages = nil
				}
			}
		}
	}()

	// 处理生产者的成功和错误消息
	go func() {
		for success := range producer.Successes() {
			log.Printf("Message is stored in topic(%s)/partition(%d)/offset(%d)\n", success.Topic, success.Partition, success.Offset)
		}
	}()

	go func() {
		for err := range producer.Errors() {
			log.Printf("Failed to send message: %v\n", err.Err)
		}
	}()

	// 启动HTTP服务器
	http.HandleFunc("/openrtb", func(w http.ResponseWriter, r *http.Request) {
		var bidRequest openrtb2.BidRequest
		if err := json.NewDecoder(r.Body).Decode(&bidRequest); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		defer r.Body.Close()

		// 提取设备信息
		deviceInfo := extractDeviceInfo(bidRequest)

		// 发送设备信息到通道
		messageChan <- &message{value: []byte(deviceInfo)}

		w.WriteHeader(http.StatusNoContent)
	})

	log.Println("Starting server at port 8080")
	if err := http.ListenAndServe(":8080", nil); err != nil {
		log.Fatalf("Could not start server: %v", err)
	}
}

func extractDeviceInfo(bidRequest openrtb2.BidRequest) string {
	// 假设bidRequest.Device包含设备信息
	device := bidRequest.Device
	app := bidRequest.App

	deviceInfo := map[string]interface{}{
		"publisher":      app.Publisher.ID,
		"bundle":         app.Bundle,
		"geo":            device.Geo.Country,
		"device_type":    device.DeviceType, // 设备类型（手机/平板等）
		"make":           device.Make,
		"model":          device.Model,
		"os":             device.OS, // 操作系统
		"osv":            device.OSV,
		"language":       device.Language,
		"connectionType": device.ConnectionType,
		"device_id":      device.IFA, // 设备ID（根据规范可选其他字段）
		"ip":             device.IP,
		"ipv6":           device.IPv6,            // IPv6地址
		"user_agent":     device.UA,              // User-Agent
		"timestamp":      time.Now().UnixMilli(), // 添加时间戳
	}

	deviceInfoJSON, err := json.Marshal(deviceInfo)
	if err != nil {
		log.Printf("Failed to marshal device info: %v", err)
		return ""
	}
	return string(deviceInfoJSON)
}

func sendMessages(producer sarama.AsyncProducer, messages []*sarama.ProducerMessage) {
	for _, msg := range messages {
		producer.Input() <- msg
		//num++
		//log.Printf("Sent %d messages to Kafka", num)
	}
}
